package com.klm.easymq.core.impl;

import com.klm.easymq.config.EasyMqProperties;
import com.klm.easymq.core.MonitorService;
import lombok.extern.slf4j.Slf4j;

import java.sql.*;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;

/**
 * 增强版SQLite监控服务实现
 * 专门用于MQ故障排查和生命周期监控
 */
@Slf4j
public class EnhancedSqliteMonitorService implements MonitorService {
    
    private final String dbPath;
    private final boolean walMode;
    private final int poolSize;
    private final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    
    public EnhancedSqliteMonitorService(EasyMqProperties.Sqlite sqliteConfig) {
        this.dbPath = sqliteConfig.getDbPath();
        this.walMode = sqliteConfig.isWalMode();
        this.poolSize = sqliteConfig.getPoolSize();
        
        initDatabase();
        log.info("Initialized Enhanced SQLite monitor service with dbPath={}, walMode={}, poolSize={}", 
                dbPath, walMode, poolSize);
    }
    
    @Override
    public void recordMessageSend(String traceId, String topic, String message, String status) {
        String sql = "INSERT OR REPLACE INTO message_records (trace_id, topic, message, status, type, create_time) VALUES (?, ?, ?, ?, ?, ?)";
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            
            stmt.setString(1, traceId);
            stmt.setString(2, topic);
            stmt.setString(3, message);
            stmt.setString(4, status);
            stmt.setString(5, "SEND");
            stmt.setString(6, LocalDateTime.now().format(formatter));
            
            stmt.executeUpdate();
            log.debug("Recorded message send: traceId={}, topic={}, status={}", traceId, topic, status);
            
        } catch (SQLException e) {
            log.error("Failed to record message send: traceId={}, topic={}", traceId, topic, e);
        }
    }
    
    @Override
    public void recordMessageConsume(String traceId, String topic, String status, String errorMsg) {
        String sql = "UPDATE message_records SET consume_status = ?, consume_time = ?, error_msg = ? WHERE trace_id = ?";
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            
            stmt.setString(1, status);
            stmt.setString(2, LocalDateTime.now().format(formatter));
            stmt.setString(3, errorMsg);
            stmt.setString(4, traceId);
            
            int updated = stmt.executeUpdate();
            if (updated == 0) {
                insertConsumeRecord(traceId, topic, status, errorMsg);
            }
            
            log.debug("Recorded message consume: traceId={}, topic={}, status={}", traceId, topic, status);
            
        } catch (SQLException e) {
            log.error("Failed to record message consume: traceId={}, topic={}", traceId, topic, e);
        }
    }
    
    @Override
    public void recordConsumerLifecycle(String consumerId, String group, String topic, String eventType, String details) {
        String sql = "INSERT INTO consumer_lifecycle (consumer_id, consumer_group, topic, event_type, details, create_time) VALUES (?, ?, ?, ?, ?, ?)";
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            
            stmt.setString(1, consumerId);
            stmt.setString(2, group);
            stmt.setString(3, topic);
            stmt.setString(4, eventType);
            stmt.setString(5, details);
            stmt.setString(6, LocalDateTime.now().format(formatter));
            
            stmt.executeUpdate();
            log.info("Recorded consumer lifecycle: consumerId={}, eventType={}, topic={}", consumerId, eventType, topic);
            
        } catch (SQLException e) {
            log.error("Failed to record consumer lifecycle: consumerId={}, eventType={}", consumerId, eventType, e);
        }
    }
    
    @Override
    public void recordConnectionStatus(String adapterType, String status, String errorMsg) {
        String sql = "INSERT INTO connection_status (adapter_type, status, error_msg, create_time) VALUES (?, ?, ?, ?)";
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            
            stmt.setString(1, adapterType);
            stmt.setString(2, status);
            stmt.setString(3, errorMsg);
            stmt.setString(4, LocalDateTime.now().format(formatter));
            
            stmt.executeUpdate();
            log.info("Recorded connection status: adapterType={}, status={}", adapterType, status);
            
        } catch (SQLException e) {
            log.error("Failed to record connection status: adapterType={}, status={}", adapterType, status, e);
        }
    }
    
    @Override
    public void recordRetryEvent(String traceId, String topic, int attempt, String errorMsg) {
        String sql = "INSERT INTO retry_events (trace_id, topic, attempt, error_msg, create_time) VALUES (?, ?, ?, ?, ?)";
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            
            stmt.setString(1, traceId);
            stmt.setString(2, topic);
            stmt.setInt(3, attempt);
            stmt.setString(4, errorMsg);
            stmt.setString(5, LocalDateTime.now().format(formatter));
            
            stmt.executeUpdate();
            log.debug("Recorded retry event: traceId={}, topic={}, attempt={}", traceId, topic, attempt);
            
        } catch (SQLException e) {
            log.error("Failed to record retry event: traceId={}, topic={}", traceId, topic, e);
        }
    }
    
    @Override
    public void recordIdempotentSkip(String traceId, String topic, String reason) {
        String sql = "INSERT INTO idempotent_skips (trace_id, topic, reason, create_time) VALUES (?, ?, ?, ?)";
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            
            stmt.setString(1, traceId);
            stmt.setString(2, topic);
            stmt.setString(3, reason);
            stmt.setString(4, LocalDateTime.now().format(formatter));
            
            stmt.executeUpdate();
            log.debug("Recorded idempotent skip: traceId={}, topic={}, reason={}", traceId, topic, reason);
            
        } catch (SQLException e) {
            log.error("Failed to record idempotent skip: traceId={}, topic={}", traceId, topic, e);
        }
    }
    
    @Override
    public List<Map<String, Object>> queryByTraceId(String traceId) {
        String sql = "SELECT * FROM message_records WHERE trace_id = ?";
        List<Map<String, Object>> results = new ArrayList<>();
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            
            stmt.setString(1, traceId);
            ResultSet rs = stmt.executeQuery();
            
            while (rs.next()) {
                Map<String, Object> record = new HashMap<>();
                record.put("traceId", rs.getString("trace_id"));
                record.put("topic", rs.getString("topic"));
                record.put("message", rs.getString("message"));
                record.put("status", rs.getString("status"));
                record.put("consumeStatus", rs.getString("consume_status"));
                record.put("type", rs.getString("type"));
                record.put("createTime", rs.getString("create_time"));
                record.put("consumeTime", rs.getString("consume_time"));
                record.put("errorMsg", rs.getString("error_msg"));
                results.add(record);
            }
            
        } catch (SQLException e) {
            log.error("Failed to query by traceId: {}", traceId, e);
        }
        
        return results;
    }
    
    @Override
    public List<Map<String, Object>> queryFailedMessages(String topic, String startTime, String endTime, int limit) {
        StringBuilder sql = new StringBuilder();
        sql.append("SELECT * FROM message_records WHERE (status = 'FAILED' OR consume_status = 'FAILED')");
        List<Object> params = new ArrayList<>();
        
        if (topic != null && !topic.isEmpty()) {
            sql.append(" AND topic = ?");
            params.add(topic);
        }
        
        if (startTime != null && !startTime.isEmpty()) {
            sql.append(" AND create_time >= ?");
            params.add(startTime);
        }
        
        if (endTime != null && !endTime.isEmpty()) {
            sql.append(" AND create_time <= ?");
            params.add(endTime);
        }
        
        sql.append(" ORDER BY create_time DESC LIMIT ?");
        params.add(limit);
        
        List<Map<String, Object>> results = new ArrayList<>();
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql.toString())) {
            
            for (int i = 0; i < params.size(); i++) {
                stmt.setObject(i + 1, params.get(i));
            }
            
            ResultSet rs = stmt.executeQuery();
            while (rs.next()) {
                Map<String, Object> record = new HashMap<>();
                record.put("traceId", rs.getString("trace_id"));
                record.put("topic", rs.getString("topic"));
                record.put("message", rs.getString("message"));
                record.put("status", rs.getString("status"));
                record.put("consumeStatus", rs.getString("consume_status"));
                record.put("errorMsg", rs.getString("error_msg"));
                record.put("createTime", rs.getString("create_time"));
                record.put("consumeTime", rs.getString("consume_time"));
                results.add(record);
            }
            
        } catch (SQLException e) {
            log.error("Failed to query failed messages", e);
        }
        
        return results;
    }
    
    @Override
    public List<Map<String, Object>> queryConsumerLifecycle(String consumerId, String group, String startTime, String endTime) {
        StringBuilder sql = new StringBuilder();
        sql.append("SELECT * FROM consumer_lifecycle WHERE 1=1");
        List<Object> params = new ArrayList<>();
        
        if (consumerId != null && !consumerId.isEmpty()) {
            sql.append(" AND consumer_id = ?");
            params.add(consumerId);
        }
        
        if (group != null && !group.isEmpty()) {
            sql.append(" AND consumer_group = ?");
            params.add(group);
        }
        
        if (startTime != null && !startTime.isEmpty()) {
            sql.append(" AND create_time >= ?");
            params.add(startTime);
        }
        
        if (endTime != null && !endTime.isEmpty()) {
            sql.append(" AND create_time <= ?");
            params.add(endTime);
        }
        
        sql.append(" ORDER BY create_time DESC");
        
        List<Map<String, Object>> results = new ArrayList<>();
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql.toString())) {
            
            for (int i = 0; i < params.size(); i++) {
                stmt.setObject(i + 1, params.get(i));
            }
            
            ResultSet rs = stmt.executeQuery();
            while (rs.next()) {
                Map<String, Object> record = new HashMap<>();
                record.put("consumerId", rs.getString("consumer_id"));
                record.put("consumerGroup", rs.getString("consumer_group"));
                record.put("topic", rs.getString("topic"));
                record.put("eventType", rs.getString("event_type"));
                record.put("details", rs.getString("details"));
                record.put("createTime", rs.getString("create_time"));
                results.add(record);
            }
            
        } catch (SQLException e) {
            log.error("Failed to query consumer lifecycle", e);
        }
        
        return results;
    }
    
    @Override
    public List<Map<String, Object>> queryConnectionHistory(String adapterType, String startTime, String endTime) {
        StringBuilder sql = new StringBuilder();
        sql.append("SELECT * FROM connection_status WHERE 1=1");
        List<Object> params = new ArrayList<>();
        
        if (adapterType != null && !adapterType.isEmpty()) {
            sql.append(" AND adapter_type = ?");
            params.add(adapterType);
        }
        
        if (startTime != null && !startTime.isEmpty()) {
            sql.append(" AND create_time >= ?");
            params.add(startTime);
        }
        
        if (endTime != null && !endTime.isEmpty()) {
            sql.append(" AND create_time <= ?");
            params.add(endTime);
        }
        
        sql.append(" ORDER BY create_time DESC");
        
        List<Map<String, Object>> results = new ArrayList<>();
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql.toString())) {
            
            for (int i = 0; i < params.size(); i++) {
                stmt.setObject(i + 1, params.get(i));
            }
            
            ResultSet rs = stmt.executeQuery();
            while (rs.next()) {
                Map<String, Object> record = new HashMap<>();
                record.put("adapterType", rs.getString("adapter_type"));
                record.put("status", rs.getString("status"));
                record.put("errorMsg", rs.getString("error_msg"));
                record.put("createTime", rs.getString("create_time"));
                results.add(record);
            }
            
        } catch (SQLException e) {
            log.error("Failed to query connection history", e);
        }
        
        return results;
    }
    
    @Override
    public Map<String, Object> getTopicStats(String topic) {
        String sql = "SELECT " +
                "COUNT(*) as total, " +
                "SUM(CASE WHEN status = 'SUCCESS' THEN 1 ELSE 0 END) as sendSuccess, " +
                "SUM(CASE WHEN status = 'FAILED' THEN 1 ELSE 0 END) as sendFailed, " +
                "SUM(CASE WHEN consume_status = 'SUCCESS' THEN 1 ELSE 0 END) as consumeSuccess, " +
                "SUM(CASE WHEN consume_status = 'FAILED' THEN 1 ELSE 0 END) as consumeFailed " +
                "FROM message_records WHERE topic = ?";
        
        Map<String, Object> stats = new HashMap<>();
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            
            stmt.setString(1, topic);
            ResultSet rs = stmt.executeQuery();
            
            if (rs.next()) {
                stats.put("total", rs.getInt("total"));
                stats.put("sendSuccess", rs.getInt("sendSuccess"));
                stats.put("sendFailed", rs.getInt("sendFailed"));
                stats.put("consumeSuccess", rs.getInt("consumeSuccess"));
                stats.put("consumeFailed", rs.getInt("consumeFailed"));
                stats.put("lastUpdate", LocalDateTime.now().format(formatter));
            }
            
        } catch (SQLException e) {
            log.error("Failed to get topic stats: {}", topic, e);
        }
        
        return stats;
    }
    
    @Override
    public Map<String, Object> getFailureStats(String startTime, String endTime) {
        StringBuilder sql = new StringBuilder();
        sql.append("SELECT " +
                "COUNT(*) as totalFailures, " +
                "COUNT(DISTINCT topic) as failedTopics, " +
                "COUNT(DISTINCT trace_id) as failedMessages, " +
                "SUM(CASE WHEN status = 'FAILED' THEN 1 ELSE 0 END) as sendFailures, " +
                "SUM(CASE WHEN consume_status = 'FAILED' THEN 1 ELSE 0 END) as consumeFailures " +
                "FROM message_records WHERE (status = 'FAILED' OR consume_status = 'FAILED')");
        
        List<Object> params = new ArrayList<>();
        
        if (startTime != null && !startTime.isEmpty()) {
            sql.append(" AND create_time >= ?");
            params.add(startTime);
        }
        
        if (endTime != null && !endTime.isEmpty()) {
            sql.append(" AND create_time <= ?");
            params.add(endTime);
        }
        
        Map<String, Object> stats = new HashMap<>();
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql.toString())) {
            
            for (int i = 0; i < params.size(); i++) {
                stmt.setObject(i + 1, params.get(i));
            }
            
            ResultSet rs = stmt.executeQuery();
            if (rs.next()) {
                stats.put("totalFailures", rs.getInt("totalFailures"));
                stats.put("failedTopics", rs.getInt("failedTopics"));
                stats.put("failedMessages", rs.getInt("failedMessages"));
                stats.put("sendFailures", rs.getInt("sendFailures"));
                stats.put("consumeFailures", rs.getInt("consumeFailures"));
                stats.put("lastUpdate", LocalDateTime.now().format(formatter));
            }
            
        } catch (SQLException e) {
            log.error("Failed to get failure stats", e);
        }
        
        return stats;
    }
    
    @Override
    public void cleanupExpiredData(int retentionDays) {
        String sql = "DELETE FROM message_records WHERE create_time < datetime('now', '-" + retentionDays + " days')";
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            
            int deleted = stmt.executeUpdate();
            log.info("Cleaned up {} expired records (older than {} days)", deleted, retentionDays);
            
        } catch (SQLException e) {
            log.error("Failed to cleanup expired data", e);
        }
    }
    
    @Override
    public String getType() {
        return "enhanced-sqlite";
    }
    
    /**
     * 初始化数据库
     */
    private void initDatabase() {
        // 消息记录表
        String createMessageRecordsTable = "CREATE TABLE IF NOT EXISTS message_records (" +
                "trace_id TEXT PRIMARY KEY, " +
                "topic TEXT NOT NULL, " +
                "message TEXT, " +
                "status TEXT, " +
                "consume_status TEXT, " +
                "type TEXT, " +
                "create_time TEXT, " +
                "consume_time TEXT, " +
                "error_msg TEXT" +
                ")";
        
        // 消费者生命周期表
        String createConsumerLifecycleTable = "CREATE TABLE IF NOT EXISTS consumer_lifecycle (" +
                "id INTEGER PRIMARY KEY AUTOINCREMENT, " +
                "consumer_id TEXT NOT NULL, " +
                "consumer_group TEXT, " +
                "topic TEXT, " +
                "event_type TEXT NOT NULL, " +
                "details TEXT, " +
                "create_time TEXT" +
                ")";
        
        // 连接状态表
        String createConnectionStatusTable = "CREATE TABLE IF NOT EXISTS connection_status (" +
                "id INTEGER PRIMARY KEY AUTOINCREMENT, " +
                "adapter_type TEXT NOT NULL, " +
                "status TEXT NOT NULL, " +
                "error_msg TEXT, " +
                "create_time TEXT" +
                ")";
        
        // 重试事件表
        String createRetryEventsTable = "CREATE TABLE IF NOT EXISTS retry_events (" +
                "id INTEGER PRIMARY KEY AUTOINCREMENT, " +
                "trace_id TEXT NOT NULL, " +
                "topic TEXT, " +
                "attempt INTEGER, " +
                "error_msg TEXT, " +
                "create_time TEXT" +
                ")";
        
        // 幂等跳过表
        String createIdempotentSkipsTable = "CREATE TABLE IF NOT EXISTS idempotent_skips (" +
                "id INTEGER PRIMARY KEY AUTOINCREMENT, " +
                "trace_id TEXT NOT NULL, " +
                "topic TEXT, " +
                "reason TEXT, " +
                "create_time TEXT" +
                ")";
        
        try (Connection conn = getConnection();
             Statement stmt = conn.createStatement()) {
            
            // 创建表
            stmt.execute(createMessageRecordsTable);
            stmt.execute(createConsumerLifecycleTable);
            stmt.execute(createConnectionStatusTable);
            stmt.execute(createRetryEventsTable);
            stmt.execute(createIdempotentSkipsTable);
            
            // 创建索引
            stmt.execute("CREATE INDEX IF NOT EXISTS idx_topic ON message_records(topic)");
            stmt.execute("CREATE INDEX IF NOT EXISTS idx_create_time ON message_records(create_time)");
            stmt.execute("CREATE INDEX IF NOT EXISTS idx_consumer_id ON consumer_lifecycle(consumer_id)");
            stmt.execute("CREATE INDEX IF NOT EXISTS idx_consumer_group ON consumer_lifecycle(consumer_group)");
            stmt.execute("CREATE INDEX IF NOT EXISTS idx_adapter_type ON connection_status(adapter_type)");
            stmt.execute("CREATE INDEX IF NOT EXISTS idx_trace_id ON retry_events(trace_id)");
            stmt.execute("CREATE INDEX IF NOT EXISTS idx_skip_trace_id ON idempotent_skips(trace_id)");
            
            log.info("Enhanced SQLite database initialized successfully with fault diagnosis tables");
            
        } catch (SQLException e) {
            log.error("Failed to initialize Enhanced SQLite database", e);
        }
    }
    
    /**
     * 获取数据库连接
     */
    private Connection getConnection() throws SQLException {
        Connection conn = DriverManager.getConnection("jdbc:sqlite:" + dbPath);
        if (walMode) {
            conn.createStatement().execute("PRAGMA journal_mode=WAL");
        }
        return conn;
    }
    
    /**
     * 插入消费记录
     */
    private void insertConsumeRecord(String traceId, String topic, String status, String errorMsg) {
        String sql = "INSERT INTO message_records (trace_id, topic, status, type, create_time, consume_status, consume_time, error_msg) VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
        
        try (Connection conn = getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            
            stmt.setString(1, traceId);
            stmt.setString(2, topic);
            stmt.setString(3, status);
            stmt.setString(4, "CONSUME");
            stmt.setString(5, LocalDateTime.now().format(formatter));
            stmt.setString(6, status);
            stmt.setString(7, LocalDateTime.now().format(formatter));
            stmt.setString(8, errorMsg);
            
            stmt.executeUpdate();
            
        } catch (SQLException e) {
            log.error("Failed to insert consume record: traceId={}", traceId, e);
        }
    }
} 